

package org.kth.dks.dks_dht;

import java.net.URL;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.log4j.Logger;
import org.kth.dks.DKSAppInterface;
import org.kth.dks.DKSCallbackInterface;
import org.kth.dks.DKSDHTInterface;
import org.kth.dks.DKSDHTVisualizationInterface;
import org.kth.dks.DKSImpl;
import org.kth.dks.DKSObject;
import org.kth.dks.DKSObjectTypes;
import org.kth.dks.dks_comm.ConnectionManager;
import org.kth.dks.dks_comm.DKSOverlayAddress;
import org.kth.dks.dks_comm.DKSRef;
import org.kth.dks.dks_exceptions.DKSIdentifierAlreadyTaken;
import org.kth.dks.dks_exceptions.DKSNodeAlreadyRegistered;
import org.kth.dks.dks_exceptions.DKSRefNoResponse;
import org.kth.dks.dks_exceptions.DKSTooManyRestartJoins;
import org.kth.dks.dks_marshal.DKSMarshal;
import org.kth.dks.dks_marshal.DKSMessage;
import org.kth.dks.dks_node.DKSNode;
import org.kth.dks.dks_node.Interval;
import org.kth.dks.util.AsyncOperation;
import org.kth.dks.util.DKSPrintTypes;
import org.kth.dks.util.MathMisc;
import org.kth.dks.util.MathMiscConstant;
import org.kth.dks.util.TimeoutException;

public class DKSDHTImpl
implements DKSDHTInterface, DKSAppInterface, DKSDHTVisualizationInterface, FailedIntervalCallbackInterface{
	private Logger log = Logger.getLogger(DKSDHTInterface.class);
	DKSCallbackInterface dksCallbacks=null;
	DHTStorage [] store=null;
	private static final long LOOKUPTIMEOUT=5000; // 5 SEC
	private static final long INSERTTIMEOUT=5000; // 5 SEC
	private static final long RESTOREITEMSTIMEOUT = 10000; // 10 SEC
	
	private static final int  DHTSUCCESS=0;
	private static final int  DHTFAILURE=0;
	private static final int  DHTNOSUCHITEM=0;
	private static final int  LOOKUPRETRYFACTOR = 2;
	
	private static final int DHTSTATUSJOINING     = 1;
	private static final int DHTSTATUSOPERATIONAL = 2;
	private static final int DHTSTATUSRESTORING   = 3;
	private static final int DHTSTATUSLEAVING     = 4;
	
	public  DKSImpl myDKSImpl=null;
	private long myId;
	
	private int f_factor = 4;
	private int lookup_factor = 4;
	
	// Statistics
	boolean insertSeen = false;
	boolean lookupSeen = false;
	
	private int status = DHTSTATUSJOINING;
	
	FailedIntervalHandler failedInterval = null;
	
	
	MathMiscConstant math = null;
	
	
	private DKSDHTCallback dhtCallback = null;
	
	
	public DKSDHTImpl(ConnectionManager cm, long nodeId, URL nodeAddress) throws DKSNodeAlreadyRegistered
	{
		store=new DHTMemoryStorage[f_factor];
		for(int i = 0 ; i < f_factor; i ++)
			store[i] = new DHTMemoryStorage();
		
		myDKSImpl=new DKSImpl(cm, nodeId, nodeAddress);
		
		myDKSImpl.setCallbackHandler(this);
		dksCallbacks=myDKSImpl.setCallbackHandler(this);
		
		setDKSCallbacks(dksCallbacks);
		myId=nodeId;
		
		math = new MathMiscConstant( DKSNode.N,2);
		
		failedInterval = new FailedIntervalHandler(this, math, f_factor, RESTOREITEMSTIMEOUT);
		
		installHandlers();
		// messages sent by route
	}
	
	public static DKSDHTImpl createDKSDiskDHT(ConnectionManager cm, DKSOverlayAddress over) throws DKSNodeAlreadyRegistered {
		return new DKSDHTImpl(cm, over, DHTDiskStorage.class);
	}
	
	public DKSDHTImpl(ConnectionManager cm, DKSOverlayAddress over) throws DKSNodeAlreadyRegistered {
		this(cm, over, DHTMemoryStorage.class);
	}
	
	
	public DKSDHTImpl(ConnectionManager cm, DKSOverlayAddress over, Class ldht) throws DKSNodeAlreadyRegistered
	{
		store=new DHTStorage[f_factor];
		try {
			for(int i = 0 ; i < f_factor; i ++)
				store[i] = (DHTStorage) ldht.newInstance();
		} catch (InstantiationException e) {
			throw new InstantiationError();
		} catch (IllegalAccessException e) {
			throw new IllegalAccessError();
		}
		
		myDKSImpl=new DKSImpl(cm, over);
		
		myDKSImpl.setCallbackHandler(this);
		dksCallbacks=myDKSImpl.setCallbackHandler(this);
		
		setDKSCallbacks(dksCallbacks);
		myId=over.getID();
		math = new MathMiscConstant( DKSNode.N,2);
		
		failedInterval = new FailedIntervalHandler(this, math, f_factor, RESTOREITEMSTIMEOUT);
		installHandlers();
		// messages sent by route
	}
	
	/**
	 * cb is the callback object
	 */
	public DKSDHTImpl(ConnectionManager cm, DKSOverlayAddress over,
			DKSDHTCallback cb)
	throws DKSNodeAlreadyRegistered
	{
		store=new DHTMemoryStorage[f_factor];
		for(int i = 0 ; i < f_factor; i ++)
			store[i] = new DHTMemoryStorage();
		
		dhtCallback = cb;
		
		myDKSImpl=new DKSImpl(cm, over);
		
		myDKSImpl.setCallbackHandler(this);
		dksCallbacks=myDKSImpl.setCallbackHandler(this);
		
		setDKSCallbacks(dksCallbacks);
		myId=over.getID();
		math = new MathMiscConstant( DKSNode.N,2);
		
		failedInterval = new FailedIntervalHandler(this, math, f_factor, RESTOREITEMSTIMEOUT);
		installHandlers();
		// messages sent by route
	}
	
	private void installHandlers() {
		DKSOverlayAddress oa = myDKSImpl.getDKSRef().getOverlayAddress();
		
		DKSMessage.addMessageTypePrefixed(AddItemMsg.NAME,
		"dks_dht.AddItemMsg");
		
		DKSMessage.addMessageTypePrefixed(RemoveItemMsg.NAME,
		"dks_dht.RemoveItemMsg");
		
		DKSMessage.addMessageTypePrefixed(ChangeItemMsg.NAME,
		"dks_dht.ChangeItemMsg");
		
		DKSMessage.addMessageTypePrefixed(GetItemsMsg.NAME,
		"dks_dht.GetItemsMsg");
		
		DKSMessage.addMessageTypePrefixed(DHTRestoreReplicasMsg.NAME,
		"dks_dht.DHTRestoreReplicasMsg");
		
		// A result message, used to indicate the result of
		// operations passed by route.
		DKSMessage.addMessageTypePrefixed(DHTResultMsg.NAME,
		"dks_dht.DHTResultMsg");
		
		// Broadcast messages
		
		
		
		
		// Out-of-band-messages.
		
		DKSMessage.addMessageTypePrefixed(RetrieveItemsMsg.NAME,
		"dks_dht.RetrieveItemsMsg");
		myDKSImpl.getDKSMarshal().addMsgHandlerPrefixed(oa,
				"dks_dht.RetrieveItemsMsg",
				"dks_dht.DKSDHTImpl",
				"retrieveItemsH",
				this);
		
		
		DKSMessage.addMessageTypePrefixed(ReplicateMsg.NAME,
		"dks_dht.ReplicateMsg");
		myDKSImpl.getDKSMarshal().addMsgHandlerPrefixed(oa,
				"dks_dht.ReplicateMsg",
				"dks_dht.DKSDHTImpl",
				"replicateH",
				this);
		
		DKSMessage.addMessageTypePrefixed(ReplicationFinishedMsg.NAME,
		"dks_dht.ReplicationFinishedMsg");
		
		myDKSImpl.getDKSMarshal().addMsgHandlerPrefixed(oa,
				"dks_dht.ReplicationFinishedMsg",
				"dks_dht.DKSDHTImpl",
				"replicationFinishedH",
				this);
		DKSMessage.addMessageTypePrefixed(DHTRestoreItemsMsg.NAME,
		"dks_dht.DHTRestoreItemsMsg");
		myDKSImpl.getDKSMarshal().addMsgHandlerPrefixed(oa,
				"dks_dht.DHTRestoreItemsMsg",
				"dks_dht.DKSDHTImpl",
				"RestoreItemsH",
				this);
		
		
	}
	
	public void create() {
		status = DHTSTATUSOPERATIONAL;
		myDKSImpl.create();
	}
	
	
	/** should be deprecated **/
	public void join(long existingnodeId, URL existingnodeAddress ) throws DKSTooManyRestartJoins, DKSIdentifierAlreadyTaken    {
		myDKSImpl.join(existingnodeId, existingnodeAddress);
	}
	
	public void join(DKSRef existingnodeAddress) throws DKSTooManyRestartJoins, DKSIdentifierAlreadyTaken, DKSRefNoResponse
	{
		myDKSImpl.join(existingnodeAddress);
	}
	
	public void logLevel(int level){
		myDKSImpl.logLevel(level);
	}
	
	/**
	 * Disconnect the local node from the network.
	 */
	public void leave(){
		myDKSImpl.leave();
	}
	
	/**
	 * Value is added to the list of bindings associated with key.
	 *
	 */
	public void addToBinding(long key, DKSObject value) {
		key = key % DKSNode.N;   // Ensure key is within right range
		AsyncOperation [] ops = new AsyncOperation[f_factor];
		for(int r = 0; r < f_factor ; r++){
			long replicaKey = associatedIdentifier(key, (r + 1));
			DKSObject dhtAddObj = new DKSObject(new AddItemMsg(key,(r+1), value).flatten());
			ops[r] = myDKSImpl.getMyDKSNode().route(replicaKey,dhtAddObj);
		}
		for(int r = 1; r <= f_factor ;r++){
			try{
				ops[r-1].waitOn(INSERTTIMEOUT);
			} catch (Exception ex) {
				log.error( "route: AsyncOperation was interrupted or cancelled");
			}
		}
	}
	
	
	
	public AsyncOperation [] addToBindingAsync(long key, DKSObject value) {
		key = key % DKSNode.N;   // Ensure key is within right range
		AsyncOperation [] ops = new AsyncOperation[f_factor];
		for(int r = 1; r <= f_factor ; r++){
			long replicaKey = associatedIdentifier(key, r);
			DKSObject dhtAddObj = new DKSObject(new AddItemMsg(key,r, value).flatten());
			ops[r-1] = myDKSImpl.getMyDKSNode().route(replicaKey,dhtAddObj);
		}
		return ops;
	}
	
	
	
	
	public DKSObject route(long identifier, DKSObject payload){
		identifier = identifier % DKSNode.N;   // Ensure key is within right range
		return myDKSImpl.route(identifier,payload);
	}
	
	public void routeAsync(long identifier, DKSObject payload){
		identifier = identifier % DKSNode.N;   // Ensure key is within right range
		myDKSImpl.routeAsync(identifier,payload);
	}
	
	public void routeAsyncFrom(long identifier, DKSObject payload, DKSRef fromNode){
		identifier = identifier % DKSNode.N;   // Ensure key is within right range
		myDKSImpl.routeAsyncFrom(identifier, payload, fromNode);
	}
	
	
	/**
	 * Value is removed from the list of bindings associated with key.
	 *
	 * Any ocurrence of value in the list is determined by comparison
	 * of the DKSObject content identifier.
	 */
	public void removeFromBinding(long key, DKSObject value) {
		key = key % DKSNode.N;   // Ensure key is within right range
		AsyncOperation [] ops = new AsyncOperation[f_factor];
		for(int r = 1; r <= f_factor ; r ++){
			long replicaKey = associatedIdentifier(key, r);
			DKSObject dhtRemoveObj = new DKSObject(new RemoveItemMsg(replicaKey,r, value).flatten());
			ops[r-1] = myDKSImpl.getMyDKSNode().route(key,dhtRemoveObj);
		}
		for(int r = 1; r <= f_factor ; r++){
			try {
				ops[r-1].waitOn();
			} catch (Exception ex) {
				log.error( "route: AsyncOperation was interrupted or cancelled");
			}
		}
		//    dht.removeItem(key, value );
	}
	
	/**
	 * All ocurrences of oldValue in the bindings associated with key
	 * are replaced by newValue.
	 *
	 * Any ocurrence of value in the list is determined by comparison
	 * of the DKSObject content identifier.
	 */
	public void changeBinding(long key, DKSObject oldValue, DKSObject newValue) {
		key = key % DKSNode.N;   // Ensure key is within right range
		AsyncOperation [] ops = new AsyncOperation[f_factor];
		for(int r = 1; r <= f_factor ; r ++){
			long replicaKey = associatedIdentifier(key, r);
			DKSObject dhtChangeObj = new DKSObject(new ChangeItemMsg(replicaKey,r, oldValue, newValue).flatten());
			ops[r-1] = myDKSImpl.getMyDKSNode().route(replicaKey,dhtChangeObj);
		}
		for(int r = 1; r <= f_factor ; r++){
			try {
				ops[r-1].waitOn();
			} catch (Exception ex) {
				log.error( "route: AsyncOperation was interrupted or cancelled");
			}
		}
	}
	
	/**
	 * Returns the list of bindings associated with key.
	 *
	 * Any ocurrence of value in the list is determined by comparison
	 * of the DKSObject content identifier.
	 */
	public DKSObject[] lookupBinding(long key) {
		key = key % DKSNode.N;   // Ensure key is within right range
	
		for(int r = 1; r <= lookup_factor ; r++){
			long replicaKey = associatedIdentifier(key, r);
			
			for(int iter = LOOKUPRETRYFACTOR ; iter >0 ; iter -- ){
				DKSObject obj =  new DKSObject(new GetItemsMsg(key).flatten());
				AsyncOperation  ops = myDKSImpl.getMyDKSNode().route(replicaKey,obj);
				
				try {
					DKSObject res = (DKSObject)ops.waitOn(LOOKUPTIMEOUT);
					DHTResultMsg rmsg = (DHTResultMsg) DKSMessage.unmarshal(res.getData());
					if(rmsg.getResult() == DHTSUCCESS && rmsg.getPayload().length > 0 ){
						return rmsg.getPayload();
					}
					else{
						iter = 0;
					}
				}
				catch (TimeoutException ex) {
					log.error( "lookupBinding timed out after "+LOOKUPTIMEOUT+" msec");
				}
				catch (Exception ex) {
					log.error( "route: AsyncOperation was interrupted or cancelled\n"+ex);
				}
			}
		}
		return new DKSObject[0];
	}
	
	
	/**
	 * Returns a list of all bindings associated with keys
	 * greater than or equal to minKey,
	 * and less than or equal to maxKey.
	 */
	public DKSObject[] lookupBinding(long minKey, long maxKey) {
		minKey = minKey % DKSNode.N;   // Ensure key is within right range
		maxKey = maxKey % DKSNode.N;   // Ensure key is within right range
		return null;
	}
	
	/**
	 * Send the message to the specified recipient node.
	 */
	public void send(DKSObject message, long recipientNodeId) {
		log.error( "This version of send is not implemented!");
		
	}
	
	/**
	 * Broadcasts the given message.
	 */
	public void send( DKSObject message, DKSRef rec) {
		
		
	}
	
	/**
	 * Broadcast a message to all nodes
	 */
	public void	broadcast(DKSObject message) {
		myDKSImpl.broadcast(message);
	}
	
	/**
	 * Broadcasts to nodes with identifiers within the rand startId and endId []
	 * @param message DKSObject message to be sent
	 * @param startId long beginning of the interval
	 * @param endId long end of the interval
	 */
	public void broadcastRestricted(DKSObject message, long startId, long endId) {
		myDKSImpl.broadcastRestricted(message, startId, endId);
	}
	
	
	/************** end of interfaces **********************/
	
	
	
	public DKSMarshal getDKSMarshal() {
		return myDKSImpl.getDKSMarshal();
	}
	
	
	public void setDKSCallbacks(DKSCallbackInterface dksImpl){
		dksCallbacks = dksImpl;
	}
	
	public static DKSObject message2obj(DKSMessage m){
		byte[] arr = m.flatten();
		
		DKSObject newObj=new DKSObject(arr);
		return newObj;
	}
	
	public void routeCallbackAsync(long identifier, DKSObject payload) {
		if(payload == null)
			log.error(
			"Error, payload null inside routeCallbackAsync in DHT!");
		
		DKSMessage dksMsg = DKSMessage.unmarshal(payload.getData());
		
		if (dhtCallback != null)
			dhtCallback.dhtRouteCallbackAsync(identifier, dksMsg);
	}
	
	public DKSObject routeCallback(long identifier, DKSObject payload){
		if(payload == null)
			log.error("Error, payload null inside routeCallback in DHT!");
		
		DKSMessage dksMsg = DKSMessage.unmarshal(payload.getData());
		
		if (status == DHTSTATUSLEAVING) 
			return message2obj(new DHTResultMsg(12, DHTFAILURE));
		
		if (dksMsg instanceof AddItemMsg){
			AddItemMsg msg = (AddItemMsg) dksMsg;
			if(msg.index-1 == 0)
				insertSeen = true;
			store[msg.index-1].insertItem(msg.key, msg.payload);
			return message2obj(new DHTResultMsg(msg.key, DHTSUCCESS));
		} else if (dksMsg instanceof RemoveItemMsg) {
			RemoveItemMsg msg = (RemoveItemMsg) dksMsg;
			
			store[msg.index-1].removeItem(msg.key, msg.payload);
			return message2obj(new DHTResultMsg(msg.key, DHTSUCCESS));
		} else if (dksMsg instanceof ChangeItemMsg) {
			ChangeItemMsg msg = (ChangeItemMsg) dksMsg;
			
			store[msg.index-1].changeItem(msg.key, msg.oldPayload, msg.newPayload);
			return message2obj(new DHTResultMsg(msg.key, DHTSUCCESS));
		} else if (dksMsg instanceof RemoveFromBindingMsg) {
			RemoveFromBindingMsg msg=(RemoveFromBindingMsg) dksMsg;
			store[0].removeItem(identifier,msg.getPayload());
			DHTResultMsg res=new DHTResultMsg(msg.getMsgId(), DHTSUCCESS);
			return message2obj(res);
		} else if (dksMsg instanceof GetItemsMsg) {
			GetItemsMsg msg=(GetItemsMsg) dksMsg;
			for(int r = 1 ; r <= f_factor ; r ++){
				DKSObject[] res= store[r-1].lookupItem(msg.getKey());
				
				if(r == 1)
					lookupSeen = true;
				if(res.length > 0)
					return message2obj(new DHTResultMsg(msg.key,DHTSUCCESS,res));
			}
			return message2obj(new DHTResultMsg(msg.key,DHTSUCCESS,new DKSObject[0]));
		} else {
			// Report failure if the message was not handled
			DKSMessage result = null;
			
			if(dhtCallback != null)
				result = dhtCallback.dhtRouteCallback(identifier, dksMsg);
			
			if(result == null) {
				result = new DHTResultMsg(12, DHTFAILURE);
				log.error(
						"Message of type " +
						dksMsg.getName() +
				" not handled");
			}
			return message2obj(result);
		}
	}
	
	public void failCallback(DKSRef failed, DKSRef failedPred) {
		if (f_factor<=1) return;
		status = DHTSTATUSRESTORING;
		
		
		printGraphRecv("F" + failed, "FAILURE DETECTED"); 
		
		Interval i = new Interval(failedPred.getID(), failed.getID());
		
		log.warn("FAILURE   --- restore interval "+ i);
		failedInterval.intervalFailed(i);
	}
	
	
	public void retrieveItemsH(DKSRef sender, RetrieveItemsMsg msg){
		log.debug("JOIN: RetrieveItems from: " + sender + " predecessor: "+myDKSImpl.getNodeInfo().predecessor ); 
		printGraphRecv("JN" + sender, "RETRIEVE"); 	
		
		Interval in = new Interval( msg.getStart(),msg.getEnd());
		
		// Find the intervals that are valid and that should be passed to the newly joined
		// node
		List validIntervals = failedInterval.intervalDifference(in);
		
		
		
		List entries = new LinkedList();
		for (Iterator it=validIntervals.iterator(); it.hasNext();) {
			Interval ii = (Interval) it.next();
			entries.addAll(getItemsInInterval(ii.start, ii.end, true)) ;
		}
		
		
		// Find the intervals not yet fully restored that matches the interval that is to
		// be passed to the newly joined node.
		List lostIntervals =  failedInterval.intervalIntersection(in);
		
		log.debug("RetrieveItemsH: items: " + entries.size() + " validIntervals " + validIntervals + " lostIntervals " + lostIntervals);
		
		// remove the responsability to restore the lost intervals passed to the newly joined node.
		failedInterval.removeInterval(in);
		if (status == DHTSTATUSRESTORING && failedInterval.isEmpty())
		{
			status = DHTSTATUSOPERATIONAL;
			log.error("FAILURE: now operational, passed failed interval to joining node");
		}
		
		printGraphSend("ITMS" + sender + "_" + myDKSImpl.getDKSRef(),  "ITEMS(" + entries.size() + ")" ); 
		myDKSImpl.send(sender,new ReplicateMsg(entries, lostIntervals, ReplicateMsg.ReplicationType.JOIN));
	}
	
	public void replicateH(DKSRef sender, ReplicateMsg msg){
		printGraphRecv("ITMS"+ myDKSImpl.getDKSRef() + "_" + sender , "ITEMS(" +  msg.getTripletList().size() + ")"); 
		Iterator iter = msg.getTripletList().iterator();
		while(iter.hasNext()){
			StoreTriplet st = (StoreTriplet)iter.next();
			// At this point the predecessor still points to the leaving node, 
			// thus, we cannot do an insert check. 
			store[st.index-1].insertItem(st.key.longValue(),  st.obj);
		}
		
		for(Iterator initer = msg.getIntervals().iterator(); initer.hasNext(); ){
			failedInterval.intervalFailed((Interval) initer.next());
		}
		
		if (msg.getReplicationType()==ReplicateMsg.ReplicationType.JOIN){
			log.debug("JOIN: done, items: " + msg.getTripletList().size() + (failedInterval.isEmpty()? " OPERATIONAL " : " RESTORING INTERVAL "));
			dksCallbacks.joinCallbackReturn();
		}  else if (msg.getReplicationType()==ReplicateMsg.ReplicationType.LEAVE){
			log.debug("Predecessor left, items: " + msg.getTripletList().size() + (failedInterval.isEmpty()? " OPERATIONAL " : " RESTORING INTERVAL "));
			myDKSImpl.send(sender, new ReplicationFinishedMsg());
		}
		if(failedInterval.isEmpty()){
			status = DHTSTATUSOPERATIONAL;
		}
		else{
			status = DHTSTATUSRESTORING;
		}
	}
	
	public void replicationFinishedH(DKSRef sender, ReplicationFinishedMsg rfmsg) {
		dksCallbacks.leaveCallbackReturn();
	}
	
	
	
	public void RestoreItemsH(DKSRef sender, DHTRestoreItemsMsg msg){
		long N = myDKSImpl.addressSpace();
		int cc = msg.getCC();
		
		int id = msg.getId(); 
		
		
		printGraphRecv( "IS" + myDKSImpl.getDKSRef() + "_" + sender + "_" + id + "_" + cc, "RESTORE(" + msg.getTripletList()+ ")");
		
		log.debug("RestoreItems id: " + id + " Intervals: " + msg.getIntervals() + " items "  + msg.getTripletList().size()); 
		
		int shift = (f_factor - (cc -1)) % f_factor + 1; 
		
		for(Iterator intervals = msg.getIntervals().iterator(); intervals.hasNext(); ){
			Interval interval = (Interval) intervals.next();
			Interval complete = associatedInterval(interval,  shift);
			failedInterval.intervalRecieved(id, complete, msg.getTripletList().size());
		}
		
		for(Iterator iter = msg.getTripletList().iterator(); iter.hasNext() ; ){
			StoreTriplet st = (StoreTriplet)iter.next();
			int indx1 = (f_factor + ((st.index -1) - (cc -1))) % f_factor;
			// it can be the case that we no longer are interested in this particular key, in such a case the 
			// key is not inserted into the local store.
			if(checkInsert(st.key.longValue(), indx1 + 1))
				store[indx1].insertItem(st.key.longValue(),  st.obj);
		}
		
		if (failedInterval.isEmpty()){
			log.debug("FAILURE --, received all items, now operational");
			status = DHTSTATUSOPERATIONAL;
		}
	}
	
	
	
	public void  restoreReplicas( DHTRestoreReplicasMsg repMsg){
		Interval intervalRec = repMsg.getInterval();
		
		Interval i = math.intervalIntersection(getResponsability(), intervalRec);
		
		// This is the faked restricted broadcast, only if this node belongs to the 
		// searched for interval should the message be processed.
		if( i == null) 
			return;
		
		// if the node is either leaving or joining it does not have 
		// a correct view of the world, and should thus not answer the 
		// question. The redo mechanism of the restorereplicas mechanism
		// will handle this by resending a message for the interval this 
		// node is currently responsible for.
		
		if(status == DHTSTATUSLEAVING || status == DHTSTATUSJOINING )
			return; 
		
		List inter = failedInterval.intervalDifference(i);
		
		if(inter.size() == 0)
			return;
		
		List entries = new LinkedList();
		
		for(Iterator itr = inter.iterator() ; itr.hasNext() ; ){
			Interval intr = (Interval) itr.next();
			List items = getItemsInInterval(intr.start, intr.end , false);
			entries.addAll(items);
		}
		printGraphRecv("RST"  + repMsg.getNewNode() + "_" + repMsg.getId() + "_" + repMsg.getCC(), "RESTORE"); 
		
		log.debug("Restore match: " + i + " := " +  inter + " entries: " + entries.size()); 
		
		printGraphSend("IS" + repMsg.getNewNode() + "_" + myDKSImpl.getDKSRef() + "_" + repMsg.getId() + "_" + repMsg.getCC(), "RESTORE(" +  entries.size() + ")"); 
		
		
		myDKSImpl.send(repMsg.getNewNode(), new DHTRestoreItemsMsg(repMsg.getCC(), repMsg.getId(), entries, inter));
	}
	
	
	public void broadcastCallback(DKSObject payload) {
		if (payload.getType()==DKSObjectTypes.DKSMSG) {
			DKSMessage dksMsg = DKSMessage.unmarshal(payload.getData());
			if (dksMsg instanceof DHTRestoreReplicasMsg) {
				restoreReplicas( (DHTRestoreReplicasMsg) dksMsg);
			}
		}else if( dhtCallback != null){
			dhtCallback.dhtBroadcastCallback(payload); // TODO: change interface to send unmarshaled payload
		}
		else{
			log.error("Application level broadcast received -- no callback handler installed");
		}
	}
	
	public void joinCallback(DKSRef pred, DKSRef succ){
		log.debug("JOIN: asking "+ succ); 
		printGraphSend("JN" + myDKSImpl.getDKSRef(), "JOIN"); 
		DKSMessage smsg=new RetrieveItemsMsg(pred.getID(), myDKSImpl.getDKSRef().getID());
		myDKSImpl.send(succ,smsg);
	}
	
	public void leaveCallback(DKSRef pred, DKSRef succ){
		status = DHTSTATUSLEAVING;
		
		List entries = getItemsInInterval(pred.getID(),myDKSImpl.getDKSRef().getID() , true) ;
		
		// by passing in the nodes responsability, every interval currently managed by the
		// interval handler will be returned.
		Interval in = new Interval( pred.getID(),myDKSImpl.getDKSRef().getID());
		List lostIntervals =  failedInterval.intervalIntersection(in);
		
		// remove the responsability to restore the lost intervals passed to the newly joined node.
		failedInterval.removeInterval(in);
		
		printGraphSend("ITMS" + succ + "_" + myDKSImpl.getDKSRef(), "LEAVE(" + entries.size() + ")"); 
		myDKSImpl.send(succ,new ReplicateMsg(entries, lostIntervals, ReplicateMsg.ReplicationType.LEAVE));
	}
	
	// Replication code
	
	private long associatedIdentifier(long i, int x){
		long N = myDKSImpl.addressSpace();
		return MathMisc.modPlus(i, (long)(x - 1)* (N/f_factor), N);
	}
	
	private Interval associatedInterval(Interval i, int x){
		return new Interval(associatedIdentifier(i.start, x),
				associatedIdentifier(i.end, x));
	}
	
	
	private List getItemsInInterval(long start, long end, boolean remove){
		List entries = new LinkedList();
		
		for(int r = 1; r <= f_factor; r ++){
			Iterator keyIter = store[r-1].getEntriesIterator();
			while(keyIter.hasNext()){
				Map.Entry entry = (Map.Entry) keyIter.next();
				Long key = (Long) entry.getKey();
				long id = associatedIdentifier(key.longValue(), r); 
				if(MathMisc.belongsTo(id, start, end, DKSNode.N)){
					Iterator valueIter = ((Set)entry.getValue()).iterator();
					while(valueIter.hasNext()){
						DKSObject oo = (DKSObject) valueIter.next();
						entries.add(new StoreTriplet(r, key, oo));
					}
					if (remove)
						keyIter.remove();
				}
			}
		}
		return entries;
	}
	
	
	public DHTStatistics getStatistics(){
		int numItems = 0;
		int sizeItems = 0;
		
		int numReplicationItems = 0;
		int sizeReplicationItems = 0;
		DHTStatisticsItem data = new DHTStatisticsItem(insertSeen, lookupSeen);
		insertSeen = false;
		lookupSeen = false;
		
		for(int r = 0 ; r < f_factor ; r++){
			Iterator keyIter = store[r].getEntriesIterator();
			while(keyIter.hasNext()){
				
				Map.Entry entry = (Map.Entry) keyIter.next();
				
				if(r==0) {
					
					numItems++;
					
				} else{
					numReplicationItems++;
				}
				data.newEntry((Long)entry.getKey(), r, null);
				
				for(Iterator valueIter = ((Set)entry.getValue()).iterator();valueIter.hasNext() ; ){
					DKSObject oo = (DKSObject) valueIter.next();
					if(r==0) {
						sizeItems += oo.getData().length;
						//	data.newValue(oo.getData());
					} else
						sizeReplicationItems += oo.getData().length;
				}
			}
		}
		return new DHTStatistics(numItems, sizeItems,
				numReplicationItems, sizeReplicationItems,
				data);
	}
	
	public String getDKSURL() {
		return myDKSImpl.getDKSURL();
	}
	
	
	public long getMyId(){
		return myId;
	}
	
	public long getPredecessorId(){
		return myDKSImpl.getNodeInfo().predecessor.getID();
	}
	
	
	public long getKeyRange(){
		return DKSNode.N;
	}
	
	public DKSRef findResponsible(long id) {
		id = id % DKSNode.N;   // Ensure key is within right range
		return myDKSImpl.findResponsible(id);
	}
	
	public void unregisterNode() {
		myDKSImpl.unregisterNode();
	}
	
	public DKSDHTImpl getDHT() {
		return this;
	}
	
	
	
	public DHTStorage [] getStorage(){
		return store;
	}
	
	
	public void sendRestoreIntervals(List intervals, int cc, int id){
		printGraphSend("RST" + myDKSImpl.getDKSRef() + "_" + id + "_" + cc ,  "RESTORE");
		for (Iterator it=intervals.iterator(); it.hasNext();) {
			Interval iv = (Interval) it.next();
			Interval i = associatedInterval( iv, cc );
			
			log.debug("Broadcasting request for " + iv + " id: " + id + " cc: " + cc + " in " + i);
			DHTRestoreReplicasMsg restMsg = new DHTRestoreReplicasMsg(myDKSImpl.getDKSRef(), i, cc, id);
			DKSObject obj =  new DKSObject(DKSObjectTypes.DKSMSG, restMsg.flatten());
			broadcast(obj);
		}
	}
	
	private Interval getResponsability(){
		long predecessor = getPredecessorId(); 
		return new Interval(predecessor, myId);
	}
	
	private boolean checkInsert(long key, int indx){
		long id = associatedIdentifier(key, indx); 
		Interval rsp = getResponsability(); 
		return (math.belongsTo(id, rsp.start, rsp.end));
	}
	
	public void printGraphSend(String id, String cnt){
		// System.out.println("GRAPH_SEND#" + myDKSImpl.getMyDKSRef().getID() + "#" + myDKSImpl.getMyDKSRef().getNonce() + "#" + id + "#" + cnt); 
	}
	
	public void printGraphRecv(String id, String cnt){
		// System.out.println("GRAPH_RECV#" + myDKSImpl.getMyDKSRef().getID() + "#" + myDKSImpl.getMyDKSRef().getNonce() + "#" + id + "#" + cnt); 
	}
	
    public void send(DKSRef target, DKSMessage message) {
    	myDKSImpl.send(target, message);
    }

    /**
	 * Gets the number of times each item is replicated in the system (default 2)
	 * Postcondition: Return value must be >=1 and must divide getKeyRange()
	 * @return int replication degree
	 */
	public int getReplicationDegree() {
		return f_factor;
	}
	
	/**
	 * Gets the number of concurrent lookups made (default 2), if >=1 then several lookups are
	 * made at the same time to different replicas to increase the fault-tolerance and the response time
	 * @return int
	 */
	public int getMultipleLookupCount() {
		return lookup_factor;
	}
	
	/**
	 * Sets the number of times items are replicated in the DHT
	 * precondition: k must divide getKeyRange() (or getKeyRange() must be a multiple of k)
	 * must be invoked before join() or create().
	 * @param k int replication degree
	 */
	public void setReplicationDegree(int k) {
		f_factor = k;
	}
	
	/**
	 * Sets the number of concurrent lookups made when lookupBinding is called.
	 * must be invoked before join() or create(). if >=1 then several lookups are
	 * made at the same time to different replicas to increase the fault-tolerance and the response time
	 * @param k int
	 */
	public void setMultipleLookupCount(int k) {
		lookup_factor = k;
	}
	
    /**
     * @see DKSInterface.addMsgHandler()
     */
    public boolean addMsgHandler(DKSMessage msg, Object handlerObject, String methodName) {
    	return myDKSImpl.addMsgHandler(msg, handlerObject, methodName);
    }

    public DKSRef getDKSRef() {
    	return myDKSImpl.getDKSRef();
    }
}
